# -*- coding:utf-8 -*-
"""
@Author : 审心
@Time : 2023/4/6 15:07
"""
import csv
from dataclasses import dataclass
from io import StringIO
import time
from typing import Union, TypeVar, overload, List, Callable

import numpy as np
import pandas as pd
import pyautogui
import pyperclip
import win32gui
import xlwings as xw
from openpyxl.utils import get_column_letter
from pandas import ExcelWriter
from xlwings import Sheet, App, Book, Range, RangeRows
from common.logger.log import log_
import os

from tqdm import tqdm

# 用于Excel合并，可以带图片
from common.logger.log import log_
import win32com.client as win32


def xw_create_wps_app(visible: bool = True) -> App:
    """
    创建操作 wps excel 的 app
    :return:
    """
    xl = xw._xlwindows.COMRetryObjectWrapper(win32.Dispatch("Ket.Application"))
    # noinspection PyProtectedMember
    impl = xw._xlwindows.App(visible=True, add_book=True, xl=xl)

    # 新建一个excel（visible为是否需要可视化，True为可视化，False为后台运行）
    return xw.App(visible=visible, add_book=True, impl=impl)


def xw_delete_empty_rows(sheet: Sheet, start_row: int, end_row: int, start_column: str, end_column: str):
    """
    删除指定范围内的空行
    :param sheet: sheet 页对象
    :param start_row: 起始行号
    :param end_row: 结束行号
    :param start_column: 起始列号
    :param end_column: 结束列号
    :return: 无返回值
    """
    # 读取指定区域的值
    data = sheet.range(f'{start_column}{start_row}:{end_column}{end_row}').value

    # 将非空行写入到新的二维列表中
    new_data = [row for row in data if any(row)]

    # 清空指定区域
    sheet.range(f'{start_column}{start_row}:{end_column}{end_row}').api.Delete()

    # 将新的数据写入到指定区域
    sheet.range(f'{start_column}{start_row}').value = new_data


def xw_copy_sheet_value(source_sheet: Sheet, target_sheet: Sheet, source_data_scope: str, target_data_scope: str):
    """
    复制 源sheet页 内容 至 目标 sheet 页
    :param source_sheet: 源 sheet 页对象
    :param target_sheet: 目标 sheet 页对象
    :param source_data_scope:  源 sheet 页区域
    :param target_data_scope:  粘贴 sheet 页区域
    :return:
    """
    # 复制原表格中指定范围的数据和格式
    source_sheet.range(source_data_scope).api.Copy()
    # 将复制的数据和格式粘贴到目标表格中
    target_sheet.range(target_data_scope).value = source_sheet.range(source_data_scope).value
    target_sheet.range(target_data_scope).api.PasteSpecial(Paste=-4122)  # -4122 为保留格式的常量值
    target_sheet.range(target_data_scope).columns.autofit()
    pyperclip.copy('')


def xw_paste_content_to_sheet(sheet: Sheet, content: str, select_scope: str, clear_scope: str = None):
    """
    粘贴文本内容到 sheet 页的指定范围
    :param sheet: sheet 对象
    :param content: 文本内容
    :param select_scope: 粘贴操作前选中某范围单元表格
    :param clear_scope: 粘贴前清除sheet页 指定范围内容
    :return:
    """
    sheet.api.Activate()
    time.sleep(1)
    if clear_scope is not None:
        sheet.range(clear_scope).clear_contents()
    sheet.api.Range(select_scope).Select()
    pyperclip.copy(content)
    time.sleep(1)
    sheet.api.Paste()


def xw_autofill_sheet(sheet: Sheet, fill_scope: str, clear_scope: str = None, after_delay: int = 1):
    """
    自动快速填充(只支持前台)
    :param sheet:sheet对象
    :param fill_scope: 填充范围
    :param clear_scope: 填充前清空指定范围数据
    :param after_delay: 填充后延迟
    :return:
    """
    if clear_scope is not None:
        sheet.range(clear_scope).clear_contents()
    sheet.api.Parent.Activate()
    sheet.api.Activate()
    set_to_active(sheet)
    sheet.api.Range(fill_scope).Select()
    pyautogui.hotkey('ctrl', 'd')
    time.sleep(after_delay)


def xw_delete_rows_containing_values_v1(sheet: Sheet, condition: callable, data_range: str = None):
    """
    根据条件删除指定 sheet 指定范围 满足条件的行 (一行一行删除)
    :param sheet: sheet 对象
    :param data_range: 数据范围 默认: 全选
    :param condition: 函数这个函数接受一个当前行的一维度数组
    :return:
    """
    delete_rows = set()
    # 获取指定列的所有单元格
    if data_range:
        sht = sheet.range(data_range)
        rows = sht.rows
    else:
        sht = sheet.used_range
        rows = sht.rows

    for i, row in enumerate(rows):
        if condition([call.value for call in row]):
            delete_rows.add(rows[i].row)
    xw_delete_row_by_row_num(sht, delete_rows)


def xw_delete_row_by_row_num(sheet: Sheet, row_num: [int, List[int]]):
    """
    删除指定行号 表格
    :param sheet: sheet 页对象
    :param row_num: 行号 单个或着一个列表
    :return:
    """
    if isinstance(row_num, int):
        sheet.api.Rows(row_num).Delete()
    elif isinstance(row_num, List):
        for row in sorted(row_num, reverse=True):
            sheet.api.Rows(row).Delete()


def xw_del_sht_img_by_name(sheet: Sheet, end_name: str):
    """
    根据指定 img 对象名称 删除图像
    :param sheet: sheet 对象
    :param end_name: 图片结尾名称
    :return:
    """
    del_obj = next((p for p in sheet.pictures if p.name.endswith(end_name)), None)
    if del_obj:
        del_obj.delete()
    else:
        log_.warning(f"删除图像失败 sheet 中没有指定名称的图像 Sheet: {sheet} end_name: {end_name}")


def xw_delete_rows_containing_values_v2(sheet: Sheet, data_range: str, condition: callable = None):
    """
    根据条件删除指定 sheet 指定范围 满足条件的行 (批量删除)
    :param sheet: sheet 对象
    :param data_range: 数据范围
    :param condition: 函数这个函数接受一个当前行的一维度数组
    :return:
    """
    # 读取指定区域的值
    data = sheet.range(data_range).value
    # 清空指定区域
    sheet.range(data_range).api.Delete()
    if condition:
        new_data = [row for row in data if not condition(row)]
        # 将新的数据写入到指定区域
        sheet.range(data_range).value = new_data


def sort_sheet_by_column(select_range: xw.Range, sort_key: callable):
    """
    对指定的 sheet 范围数据进行排序
    :param select_range: 指定的数据范围
    :param sort_key: 排序关键字，需要传入一个函数，函数的参数是一行数据，函数的返回值将会用来排序
    :return:
    """
    data = [[cell.value for cell in row] for row in select_range.rows]

    # 按照指定的关键字排序
    sorted_data = sorted(data, key=sort_key)

    # 写入排序后的数据到 Excel 文件中
    select_range.value = sorted_data


def xw_get_has_content_row(sheet: Sheet, column: str = "A") -> int:
    """
    获取指定列已经有内容的行数
    :param sheet: sheet 页对象
    :param column: 列号
    :return: 行数
    """
    return sheet.range(f'{column}1').expand('table').rows.count


def xw_get_last_used_column(sheet: xw.Sheet, row: int) -> str:
    """
    获取指定行号的最后一列 字母
    :param sheet: sheet 对象
    :param row: 行号
    :return: 列名
    """
    # 获取指定行的已使用范围
    used_range = sheet.used_range.rows[row - 1].expand('right')
    # 获取最后一列的列索引
    last_column = used_range.last_cell.column
    return get_column_letter(last_column)


def xw_find_last_nonzero_row(sheet, column, content):
    """
    返回指定列最后一个不为指定内容的行号
    :param sheet: sheet页对象
    :param column: 指定列字母
    :param content: 内容
    :return: 最后一个不为指定内容的行号
    """
    last_row = xw_get_has_content_row(sheet, column)
    for row in range(1, last_row + 1):
        if sheet.range(f"{column}{row}").value == content:
            return row - 1
    return last_row


def set_to_active(target: [Sheet, Book, App]):
    """
    激活excel至最前台
    :param target: 目标 sheet 或 表格 或 应用程序
    :return:
    """
    hwnd = 0
    if isinstance(target, Sheet):
        hwnd = target.book.app.api.Hwnd
    if isinstance(target, Book):
        hwnd = target.app.api.Hwnd
    if isinstance(target, App):
        hwnd = target.api.Hwnd
    win32gui.SetForegroundWindow(hwnd)


def xw_format_ranges_as(sheet: Sheet, ranges: str, format_string: str):
    """
    将指定工作表中的指定范围格式化为指定的格式字符串
    :param sheet: 工作表对象
    :param ranges: 范围字符串，多个范围使用逗号分隔，例如'A1:B10,C1:D10'
    :param format_string: 格式字符串
    Example:
        >>> xw_format_ranges_as(sheet, 'A1:B10,C1:D10', "0.00%")
    :return:
    """
    for rng in [sheet.range(r.strip()) for r in ranges.split(',')]:
        rng.number_format = format_string


def xw_col_content_to_link(sht: Sheet, start_cell_str: str):
    """
    将指定列文本内容转为超链接
    :param sht: sheet 对象
    :param start_cell_str: 起始单元格字符串 例 : 'A2'
    :return:
    """
    for cell in sht.range(start_cell_str).expand("down"):
        cell.api.Hyperlinks.Add(Anchor=cell.api, Address=cell.value)


def xw_add_sheet_with_unique_name(book: Book, sheet_name):
    """
    向 Excel 工作簿添加一个唯一命名的工作表，并返回该工作表对象。

    Args:
        book (xlwings.Book): 要添加工作表的 Excel 工作簿对象。
        sheet_name (str): 新工作表的名称。

    Returns:
        xlwings.Sheet: 新添加的工作表对象。

    Example:
        >>> sheet = xw_add_sheet_with_unique_name(book, '新工作表')
    """
    if not isinstance(sheet_name, str):
        sheet_name = str(sheet_name)
    # 检查是否存在同名的sheet，如果有，则添加序号直到没有同名的sheet
    i = 1
    exist_names = [sheet.name for sheet in book.sheets]
    while sheet_name in exist_names:
        sheet_name = f"{sheet_name}^{i}"
        i += 1

    # 添加新sheet
    new_sheet = book.sheets.add(name=sheet_name)
    return new_sheet


def xw_append_img_to_cell(sheet: Sheet, cell: str, img_path: str):
    """
    对单元格追加图片
    :param sheet: Sheet 页独享
    :param cell: 单元格选择字符串入 A1
    :param img_path: 图片路径
    :return:
    """
    sheet.pictures.add(img_path, name=cell, left=sheet.range(cell).left,
                       top=sheet.range(cell).top, width=sheet.range(cell).width,
                       height=sheet.range(cell).height)


def xw_del_sheet(sheet, sheet_name):
    """
    删除指定名称的sheet
    :param sheet: sheet页对象
    :param sheet_name: sheet名称
    :return:
    """
    # 删除指定的工作表
    worksheet = sheet.sheets[sheet_name]
    worksheet.delete()
    log_.info(f"{sheet_name} 删除成功!")


def xw_close_all_excel():
    for app in xw.apps:
        app.quit()


def pd_paste_2_csv(content: str = None, path: str = None, str_sep='\t', data_header: int = 1) -> pd.DataFrame:
    """
    将剪切板内容或将字符串转为表格,选择是否保存为csv文件
    :param data_header: 表头选择第几行
    :param str_sep: 字符串分隔符
    :param content: 表格内容字符串
    :param path:保存csv文件路径,当为None时不保存
    :return:返回 pd.DataFrame表格
    """

    if content is not None:
        # 将字符串转化为表格
        data = pd.read_csv(StringIO(content), sep=str_sep, header=data_header)
    else:
        # 将剪切板内容转化为表格
        time.sleep(1)
        data = pd.read_clipboard(sep=str_sep, header=data_header, usecols=lambda x: 'Unnamed' not in x)
    if path is not None:
        data.to_csv(path, index=False, quoting=csv.QUOTE_ALL)
        log_.info(f'已将剪切板内容写入到{path}')

    return data


T = TypeVar('T', str, int)


# 定义个字符串与整数型的泛型
@overload
def copy_excel_section(path: str, sheet_name: str, column_index: list[str, ...],
                       row: tuple[int, int] = (None, None)) -> pd.DataFrame:
    pass


@overload
def copy_excel_section(path: str, sheet_name: str, column_index: list[int, ...],
                       row: tuple[int, int] = (None, None)) -> pd.DataFrame:
    pass


def copy_excel_section(path: str, sheet_name: str, column_index: list[Union[str, int], ...],
                       row: tuple[int, int] = (None, None)) -> pd.DataFrame:
    """
    复制excel中的指定行列数据
    column_index可以传递列名或者列索引

    :param row: 需要复制的行范围
    :param column_index: 需要复制的列名/第几列,可以传递多个
    :param sheet_name: 选择excel中需要读取的表格名
    :param path: excel路径
    :return:column_data 返回选择区域的数据
    """
    data = pd.read_excel(path, sheet_name)
    if all(isinstance(col, str) for col in column_index):
        # 如果所有元素都是字符串，则按照列名复制
        column_data = data.loc[row[0]:row[1], column_index]
    elif all(isinstance(col, int) for col in column_index):
        # 如果所有元素都是整数，则按照列索引复制
        column_data = data.iloc[row[0]:row[1], column_index]
    else:
        # 如果元组中的元素类型不一致，则抛出 TypeError 异常
        raise TypeError("元组中的所有元素必须是相同类型的")
    # 复制到剪切板
    column_data.to_clipboard(index=False, header=False)
    return column_data


def to_excel_auto_column_weight(df: pd.DataFrame, writer: ExcelWriter, sheet_name):
    """DataFrame保存为excel并自动设置列宽"""
    df.to_excel(writer, sheet_name=sheet_name, index=False)
    #  计算表头的字符宽度
    column_widths = (
        df.columns.to_series().apply(lambda x: len(x.encode('gbk'))).values
    )
    #  计算每列的最大字符宽度
    max_widths = (
        df.astype(str).applymap(lambda x: len(x.encode('gbk'))).agg(max).values
    )
    # 计算整体最大宽度
    widths = np.max([column_widths, max_widths], axis=0)
    # 设置列宽
    worksheet = writer.sheets[sheet_name]
    for i, width in enumerate(widths, 1):
        # openpyxl引擎设置字符宽度时会缩水0.5左右个字符，所以干脆+2使左右都空出一个字宽。
        worksheet.column_dimensions[get_column_letter(i)].width = width + 2
    # return worksheet


def del_english_keywords(data, column):
    """
    删除表格中的英文关键词
    :param column: 需要去除英文的列名
    :param data:表格数据
    :return:data:去除英文的表格数据
    """
    en_mask = data[column].str.contains('[a-zA-Z]')
    data = data[~en_mask]
    return data


def xlsx_merge(xlsx_files: list, sheet_name: str, output_path: str, file_name, header_parent: bool,
               row_height: int = None, column_width: int = None):
    """
    使用win32调用Excel合并支持excel中带有图片
    :param file_name: 保存文件名称
    :param column_width: 表格列宽 不设置则使用默认列宽
    :param row_height: 表格行高 不设置则使用默认行高
    :param header_parent: 是否继承第一个文件的表头
    :param output_path: 合并后输出文件路径
    :param xlsx_files: 需要合并的文件列表
    :param sheet_name: 合并后的sheet名称
    :return:
    """
    log_.info("开始合并文件")
    excel = win32.gencache.EnsureDispatch('Excel.Application')
    # 隐藏Excel窗口
    excel.Visible = False

    try:
        # 添加一个新的工作簿
        wb_merged = excel.Workbooks.Add()
        ws_merged = wb_merged.Worksheets(1)
        ws_merged.Name = sheet_name
        if row_height:
            # 设置所有行的行高
            ws_merged.Rows.RowHeight = row_height
        if column_width:
            # 设置所有列的列宽
            ws_merged.Columns.ColumnWidth = column_width
        # 遍历所有文件并将它们合并到一个工作簿中
        for i, filename in tqdm(enumerate(xlsx_files), total=len(xlsx_files)):
            # 打开原始工作簿
            wb = excel.Workbooks.Open(filename)
            ws = wb.Worksheets(1)
            # 复制工作表的内容
            if i == 0:
                # 第一次复制复制到（1，1）单元格，带表头
                ws.UsedRange.Copy(ws_merged.Cells(1, 1))
            else:
                if header_parent:
                    # 第二之后的工作表复制到已有内容的下一行， 不带表头
                    ws.Range("A2", ws.Cells(ws.UsedRange.Rows.Count, ws.UsedRange.Columns.Count)).Copy(
                        ws_merged.Cells(ws_merged.UsedRange.Rows.Count + 1, 1))
                else:
                    ws.UsedRange.Copy(ws_merged.Cells(ws_merged.UsedRange.Rows.Count + 1, 1))
        # 保存合并文件
        if not os.path.exists(output_path):
            # 如果文件夹不存在则创建
            os.makedirs(output_path)
        output_file_path = os.path.join(output_path, file_name)
        print(output_file_path, len(output_file_path))
        wb_merged.SaveAs(output_file_path)
        log_.info("合并成功")
    finally:
        # 关闭Excel应用程序
        log_.info("正在关闭Excel")
        excel.Application.Quit()
        del excel


@dataclass
class XwRowMap:
    """
    xlwings 行映射
    :param num  行号
    :param row 列范围对象
    """
    num: int
    row: Range

    def __eq__(self, other):
        if isinstance(other, XwRowMap):
            return self.num == other.num
        return False

    def __hash__(self):
        return hash(self.num)

    @staticmethod
    def builder_by_rows(range_: Range) -> List:
        """
        根据 Range 对象构建此对象列表
        :param range_: Range 对象
        :return:
        """
        return [XwRowMap(row.row, row) for row in range_.rows]

    @staticmethod
    def filter_row_map_by_col(row_map_list: List, value_index: int, compare_func: Callable,
                              ignore_row_num=None, before_condition: bool = True) -> List:
        """
        返回满足条件(compare_func = true)的 rowMap
        :param value_index: 值的列下标
        :param ignore_row_num: 不删除的行号
        :param row_map_list: 行数据映射列表
        :param compare_func: 比较对象
        :param before_condition: 执行过滤的前置条件
        :return: 过滤后的行映射列表
        """
        if ignore_row_num is None:
            ignore_row_num = []
        if not before_condition:
            return row_map_list
        result = []
        if len(row_map_list) > 0:
            result = [row for row in row_map_list
                      if compare_func(row.row.value[value_index]) or row.num in ignore_row_num]
        return result

    @staticmethod
    def filter_row_map_by_row(row_map_list: List, compare_func: Callable,
                              ignore_row_num=None, before_condition: bool = True) -> List:
        """
        返回满足条件(compare_func = true)的 rowMap
        :param ignore_row_num: 不删除的行号
        :param row_map_list: 行数据映射列表
        :param compare_func: 比较对象
        :param before_condition: 执行过滤的前置条件
        :return: 过滤后的行映射列表
        """
        if ignore_row_num is None:
            ignore_row_num = []
        if not before_condition:
            return row_map_list
        result = []
        if len(row_map_list) > 0:
            result = [row for row in row_map_list
                      if compare_func(row.row.value) or row.num in ignore_row_num]
        return result
